package com.sven;

import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.binder.BinderHeaders;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Arrays;
import java.util.List;
import java.util.UUID;


@RestController
public class SendController {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @Autowired
    private Source source;

    /**
     * 发送事务消息，预提交
     * 消息已经发送到Broker中，单还未投递出去，Consumer暂时还不能消费这条消息
     *
     * 在使用rocketMQTemplate类时,sendMessageInTransaction方法的第四个参数可以帮助我们传递对象,
     * source接口的send方法没有多余参数,所以我们利用MessageBuilder将对象信息放在消息头里面.因为setHeader只能传递字符串,
     * 所以我们将对象转换为Json字符串,然后在处理本地事务从消息头中取出来,转换回来就可以了
     */
    @RequestMapping("/send4")
    public String send4(String msg){
        String transactionId = UUID.randomUUID().toString();
        MessageBuilder builder = MessageBuilder.withPayload(msg).setHeader(RocketMQHeaders.TRANSACTION_ID, transactionId);
        Message message = builder.build();

        // 在Spring消息编程模型下，使用RocketMQ收发消息 一文中,发送消息使用的是RocketMQTemplate类.
        // 在集成了Spring Cloud Stream之后,我们可以使用Source实现消息的发送
        source.output().send(message);
        return "ok " + transactionId;
    }

    //======================================================

    //顺序消息
    @RequestMapping("/send3")
    public String send3(){
        for(int i=0; i<1000; i++){
            MessageBuilder builder = MessageBuilder.withPayload("哈哈"+i).setHeader(BinderHeaders.PARTITION_HEADER, 0);
            Message message = builder.build();
            source.output().send(message);
        }

        return "ok";
    }

    //======================================================

    @Autowired
    private OrderChannel orderChannel;

    @RequestMapping("/send2")
    public String send2(String msg){
        MessageBuilder builder = MessageBuilder.withPayload(msg);
        Message message = builder.build();
        orderChannel.output().send(message);
        return "hello " + msg;
    }

    @StreamListener(value = OrderChannel.INPUT)
    public void receive22(String msg) {
        System.out.println("=================OrderChannel.INPUT========" + msg);//哈哈哈hhh
    }

    //======================================================

    @RequestMapping("/send")
    public String send(String msg){
        MessageBuilder builder = MessageBuilder.withPayload(msg);
        Message message = builder.build();
        source.output().send(message);
        return "hello " + msg;
    }

    //使用@StreamListener来监听消息
/*    @StreamListener(value = Sink.INPUT)
    public void receive(Message message) {
        System.out.println("=================Sink.INPUT========" + message.getPayload().toString());//哈哈哈hhh
    }*/

    @StreamListener(value = Sink.INPUT)
    public void receive2(String msg) {
        System.out.println("=================Sink.INPUT========" + msg);//哈哈哈hhh
    }



}
